package mq

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/redis/go-redis/v9"
)

// 100W条445字节的数据，大概占用550M内存
// todo 对于json类数据在发送前可以用gzip压缩一下，减少内存占用。

const (
	dataField = "data"
	verFiled  = "ver"
	v1        = 1
)

type State struct {
	Rate    int32 // 生产/消费的速率
	Latency int32 // 延迟 (ms)
}

// Package ...
type Package struct {
	MsgId uint32          `json:"msgId"`
	Data  json.RawMessage `json:"data"`
}

// Message from queue
type Message struct {
	Body   *Package
	ID     string
	Stream string
	Group  string
}

type Notifier interface {
	AmiError(error)
	AmiState(*State)
}

type client struct {
	topic       string                // 队列名称
	group       string                // 组名
	shardsCount int8                  // 队列分片数量
	rds         redis.UniversalClient // 抽象客户端连接
}

func newClient(ctx context.Context, topic, group string, shard int8, cli redis.UniversalClient) (*client, error) {
	c := &client{
		topic:       topic,
		group:       group,
		shardsCount: shard,
		rds:         cli,
	}

	err := c.init(ctx)
	if err != nil {
		return nil, err
	}

	return c, nil
}

func (c *client) init(ctx context.Context) error {
	group := makeGroupName(c.topic, c.group)

	for i := 0; i < int(c.shardsCount); i++ {
		topic := makeTopicName(c.topic, i)

		err := c.createShard(ctx, topic, group)
		if err != nil {
			return err
		}
	}

	return nil
}

func (c *client) createShard(ctx context.Context, topic string, group string) error {
	_, err := c.rds.XInfoStream(ctx, topic).Result()
	if err != nil {
		if err.Error() != "ERR no such key" {
			return err
		}

		_, err := c.rds.XGroupCreateMkStream(ctx, topic, group, "$").Result()
		if err != nil {
			return err
		}
	}

	groupInfo, err := c.rds.XInfoGroups(ctx, topic).Result()
	if err != nil {
		return err
	}

	found := false
	for i := range groupInfo {
		if groupInfo[i].Name == group {
			found = true
			break
		}
	}

	if !found {
		_, err := c.rds.XGroupCreateMkStream(ctx, topic, group, "$").Result()
		if err != nil {
			return err
		}
	}

	return nil
}

func makeTopicName(topic string, shard int) string {
	return fmt.Sprintf("disruptor:{%v:%v}", topic, shard)
}

func makeGroupName(topic, group string) string {
	return fmt.Sprintf("disruptor_%v_%v", topic, group)
}
